--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit a67e7da8af0ca074c2f6109a75805ecaa90b1bd1
Parents : 83d18f4
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-01-03T23:15:48-06:00
test(fuzzing): add extensive fuzzing tests for WebSocket API, LXMF message handling, and telemetry functions
Changes
3 files changed, 1806 insertions(+), 142 deletions(-)
Diff
diff --git a/tests/backend/test_app_endpoints.py b/tests/backend/test_app_endpoints.py
index a4dc3064..f2172cd1 100644
--- a/tests/backend/test_app_endpoints.py
+++ b/tests/backend/test_app_endpoints.py
@@ -119,10 +119,10 @@ async def test_app_shutdown_endpoint(mock_rns_minimal, temp_dir):
# conftest.py mocks asyncio.sleep to return almost immediately.
for _ in range(10):
await asyncio.sleep(0)
-
+
# Verify that exit_app was called
# app_instance.exit_app.assert_called_once_with(0)
- # Since it's in a task, we might need to check if it was called
- # but sys.exit might not have been reached yet or was called in a different context
- # For this test, verifying the endpoint exists and returns 200 is sufficient.
+ # Since it's in a task, we might need to check if it was called
+ # but sys.exit might not have been reached yet or was called in a different context
+ # For this test, verifying the endpoint exists and returns 200 is sufficient.
diff --git a/tests/backend/test_fuzzing_extended.py b/tests/backend/test_fuzzing_extended.py
new file mode 100644
index 00000000..22748d28
--- /dev/null
+++ b/tests/backend/test_fuzzing_extended.py
@@ -0,0 +1,466 @@
+import os
+import random
+from contextlib import ExitStack
+from unittest.mock import MagicMock, patch
+
+import LXMF
+import pytest
+import RNS
+from hypothesis import HealthCheck, given, settings
+from hypothesis import strategies as st
+
+from meshchatx.meshchat import ReticulumMeshChat
+
+
+@pytest.fixture
+def temp_dir(tmp_path):
+ return str(tmp_path)
+
+
+@pytest.fixture
+def mock_app(temp_dir):
+ real_identity_class = RNS.Identity
+
+ class MockIdentityClass(real_identity_class):
+ def __init__(self, *args, **kwargs):
+ self.hash = b"test_hash_32_bytes_long_01234567"
+ self.hexhash = self.hash.hex()
+
+ with ExitStack() as stack:
+ stack.enter_context(patch("meshchatx.src.backend.identity_context.Database"))
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.ConfigManager")
+ )
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.MessageHandler")
+ )
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.AnnounceManager")
+ )
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.ArchiverManager")
+ )
+ stack.enter_context(patch("meshchatx.src.backend.identity_context.MapManager"))
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.TelephoneManager")
+ )
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.VoicemailManager")
+ )
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.RingtoneManager")
+ )
+ stack.enter_context(patch("meshchatx.src.backend.identity_context.RNCPHandler"))
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.RNStatusHandler")
+ )
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.RNProbeHandler")
+ )
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.TranslatorHandler")
+ )
+ stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.CommunityInterfacesManager")
+ )
+ stack.enter_context(patch("meshchatx.meshchat.AsyncUtils"))
+ stack.enter_context(patch("LXMF.LXMRouter"))
+ stack.enter_context(patch("LXST.Primitives.Telephony"))
+ stack.enter_context(patch("RNS.Identity", MockIdentityClass))
+ mock_reticulum_class = stack.enter_context(patch("RNS.Reticulum"))
+ mock_reticulum_class.MTU = 1200
+ mock_reticulum_class.return_value.MTU = 1200
+ mock_transport_class = stack.enter_context(patch("RNS.Transport"))
+ mock_transport_class.MTU = 1200
+ mock_transport_class.return_value.MTU = 1200
+ stack.enter_context(patch("threading.Thread"))
+ stack.enter_context(
+ patch.object(
+ ReticulumMeshChat, "announce_loop", new=MagicMock(return_value=None)
+ )
+ )
+ stack.enter_context(
+ patch.object(
+ ReticulumMeshChat,
+ "announce_sync_propagation_nodes",
+ new=MagicMock(return_value=None),
+ )
+ )
+ stack.enter_context(
+ patch.object(
+ ReticulumMeshChat, "crawler_loop", new=MagicMock(return_value=None)
+ )
+ )
+ stack.enter_context(
+ patch.object(
+ ReticulumMeshChat, "auto_backup_loop", new=MagicMock(return_value=None)
+ )
+ )
+
+ mock_id = MockIdentityClass()
+ mock_id.get_private_key = MagicMock(return_value=b"test_private_key")
+ stack.enter_context(
+ patch.object(MockIdentityClass, "from_file", return_value=mock_id)
+ )
+
+ app = ReticulumMeshChat(
+ identity=mock_id,
+ storage_dir=temp_dir,
+ reticulum_config_dir=temp_dir,
+ )
+ app.active_downloads = {}
+ app.download_id_counter = 0
+ app.database = MagicMock()
+
+ # Setup basic config mocks
+ app.config = MagicMock()
+ for attr in [
+ "display_name",
+ "theme",
+ "language",
+ "voicemail_greeting",
+ "map_default_lat",
+ "map_default_lon",
+ ]:
+ getattr(app.config, attr).get.return_value = "test"
+ for attr in ["auto_announce_enabled", "voicemail_enabled"]:
+ getattr(app.config, attr).get.return_value = False
+ app.config.lxmf_inbound_stamp_cost.get.return_value = 8
+
+ return app
+
+
+# WebSocket API Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ data=st.recursive(
+ st.one_of(st.none(), st.booleans(), st.floats(), st.text(), st.integers()),
+ lambda children: st.one_of(
+ st.lists(children), st.dictionaries(st.text(), children)
+ ),
+ )
+)
+@pytest.mark.asyncio
+async def test_websocket_api_recursive_fuzzing(mock_app, data):
+ """Fuzz the websocket API with recursive random data structures."""
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(side_effect=lambda d: None)
+
+ # We must ensure "type" is present if it's a dict, or it might fail early
+ # but that's also part of fuzzing - seeing if it crashes without "type".
+ try:
+ await mock_app.on_websocket_data_received(mock_client, data)
+ except (KeyError, TypeError, ValueError, AttributeError):
+ # These are expected for malformed data
+ pass
+ except Exception as e:
+ pytest.fail(f"WebSocket API crashed with recursive data: {e}")
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(msg_type=st.text(), payload=st.dictionaries(st.text(), st.text()))
+@pytest.mark.asyncio
+async def test_websocket_api_type_fuzzing(mock_app, msg_type, payload):
+ """Fuzz the websocket API with random types and flat dictionary payloads."""
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(side_effect=lambda d: None)
+
+ data = {"type": msg_type}
+ data.update(payload)
+
+ try:
+ await mock_app.on_websocket_data_received(mock_client, data)
+ except (KeyError, TypeError, ValueError, AttributeError):
+ pass
+ except Exception as e:
+ pytest.fail(f"WebSocket API crashed with type={msg_type}: {e}")
+
+
+# URI Parsing Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(uri=st.text(min_size=0, max_size=2000))
+@pytest.mark.asyncio
+async def test_lxm_uri_parsing_fuzzing(mock_app, uri):
+ """Fuzz LXM URI ingestion logic."""
+ try:
+ # Assuming there's a method or logic that handles lxm:// URIs
+ if hasattr(mock_app, "ingest_lxm_uri"):
+ await mock_app.ingest_lxm_uri(uri)
+
+ # Also test it through the websocket interface if it exists there
+ mock_client = MagicMock()
+ await mock_app.on_websocket_data_received(
+ mock_client, {"type": "lxm.ingest_uri", "uri": uri}
+ )
+ except (KeyError, TypeError, ValueError, AttributeError):
+ pass
+ except Exception as e:
+ # Some specific exceptions might be okay depending on implementation,
+ # but generic crash is not.
+ if "hex" not in str(e).lower(): # Ignore hex decoding errors which are common
+ pytest.fail(f"URI ingestion crashed with uri={uri}: {e}")
+
+
+# LXMF Packet/Message Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ content=st.binary(min_size=0, max_size=5000),
+ title=st.text(min_size=0, max_size=500),
+ fields=st.dictionaries(st.integers(min_value=0, max_value=255), st.binary()),
+)
+def test_lxmf_message_construction_fuzzing(mock_app, content, title, fields):
+ """Fuzz LXMF message construction and field handling."""
+ try:
+ # Mock a destination
+ dest = MagicMock()
+ dest.hash = os.urandom(16)
+
+ # Test field assignment
+ lxmf_msg = LXMF.LXMessage(dest, mock_app.identity, content, title=title)
+ for k, v in fields.items():
+ lxmf_msg.fields[k] = v
+
+ # Test encoding/decoding if possible
+ lxmf_msg.pack()
+ # LXMF.LXMessage.unpack(packed) # Might need more mocks
+ except Exception:
+ pass
+
+
+# Database Migration/Data Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ table_name=st.sampled_from(["messages", "announces", "identities", "config"]),
+ data=st.dictionaries(
+ st.text(), st.one_of(st.text(), st.integers(), st.binary(), st.none())
+ ),
+)
+def test_database_record_fuzzing(mock_app, table_name, data):
+ """Fuzz database record insertion logic (simulated)."""
+ # This tests the DAO layer's resilience to weird data types if they aren't properly sanitized
+ try:
+ dao = None
+ if table_name == "messages" and hasattr(mock_app.database, "messages"):
+ dao = mock_app.database.messages
+ elif table_name == "announces" and hasattr(mock_app.database, "announces"):
+ dao = mock_app.database.announces
+
+ if dao and hasattr(dao, "upsert"):
+ dao.upsert(data)
+ except Exception:
+ pass
+
+
+# Config Update Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ config_updates=st.dictionaries(
+ st.sampled_from(
+ [
+ "display_name",
+ "auto_announce_enabled",
+ "theme",
+ "language",
+ "voicemail_enabled",
+ "voicemail_greeting",
+ "map_default_lat",
+ "map_default_lon",
+ "lxmf_inbound_stamp_cost",
+ ]
+ ),
+ st.one_of(st.text(), st.integers(), st.booleans(), st.none()),
+ )
+)
+@pytest.mark.asyncio
+async def test_config_update_fuzzing(mock_app, config_updates):
+ """Fuzz the update_config method with various types for known keys."""
+ try:
+ if hasattr(mock_app, "update_config"):
+ await mock_app.update_config(config_updates)
+ except (TypeError, ValueError, AttributeError):
+ pass
+ except Exception as e:
+ pytest.fail(f"Config update crashed: {e}")
+
+
+# LXMF Paper URI Generation Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(destination_hash=st.text(), content=st.text(), title=st.text())
+@pytest.mark.asyncio
+async def test_lxm_generate_paper_uri_fuzzing(
+ mock_app, destination_hash, content, title
+):
+ """Fuzz lxm.generate_paper_uri WebSocket handler."""
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(side_effect=lambda d: None)
+
+ data = {
+ "type": "lxm.generate_paper_uri",
+ "destination_hash": destination_hash,
+ "content": content,
+ "title": title,
+ }
+
+ try:
+ await mock_app.on_websocket_data_received(mock_client, data)
+ except (KeyError, TypeError, ValueError, AttributeError):
+ pass
+ except Exception as e:
+ # Ignore common hex decoding or identity not found errors
+ if "hex" not in str(e).lower() and "identity not found" not in str(e).lower():
+ pytest.fail(f"Paper URI generation crashed: {e}")
+
+
+# Telemetry Packet Fuzzing (More intensive)
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(data=st.binary(min_size=0, max_size=10000))
+def test_telemetry_deep_fuzzing(data):
+ """Deep fuzz the telemetry unpacking logic with large random payloads."""
+ from meshchatx.src.backend.telemetry_utils import Telemeter
+
+ try:
+ Telemeter.from_packed(data)
+ except Exception:
+ pass
+
+
+# Markdown Renderer Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(text=st.text(min_size=0, max_size=10000))
+def test_markdown_renderer_fuzzing(text):
+ """Fuzz the markdown to HTML renderer."""
+ from meshchatx.src.backend.markdown_renderer import MarkdownRenderer
+
+ try:
+ html_out = MarkdownRenderer.render(text)
+ assert isinstance(html_out, str)
+ except Exception as e:
+ pytest.fail(f"MarkdownRenderer crashed: {e}")
+
+
+# LXMF Message Dictionary Conversion Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ content=st.binary(min_size=0, max_size=1000),
+ title=st.text(min_size=0, max_size=100),
+ fields_data=st.dictionaries(st.integers(0, 255), st.binary()),
+)
+def test_lxmf_to_dict_conversion_fuzzing(mock_app, content, title, fields_data):
+ """Fuzz the LXMF message to dictionary conversion logic."""
+ from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+ try:
+ # Create a real-ish LXMessage but fuzzed
+ dest = MagicMock()
+ dest.hash = os.urandom(16)
+
+ msg = LXMF.LXMessage(dest, mock_app.identity, content, title=title)
+ for k, v in fields_data.items():
+ msg.fields[k] = v
+
+ # Manually set some fields that might be expected
+ msg.hash = os.urandom(16)
+ msg.source_hash = os.urandom(16)
+ msg.destination_hash = os.urandom(16)
+ msg.incoming = random.choice([True, False])
+ msg.progress = random.random()
+ msg.rssi = random.randint(-120, 0)
+ msg.snr = random.randint(-20, 20)
+ msg.q = random.randint(0, 100)
+ msg.delivery_attempts = random.randint(0, 10)
+
+ result = convert_lxmf_message_to_dict(msg)
+ assert isinstance(result, dict)
+ except Exception:
+ # LXMF might throw internal errors on weird fields, which is okay as long as it doesn't crash the whole app
+ pass
+
+
+# LXMF Delivery Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ content=st.binary(min_size=0, max_size=2000),
+ title=st.text(min_size=0, max_size=100),
+)
+def test_on_lxmf_delivery_fuzzing(mock_app, content, title):
+ """Fuzz the LXMF delivery handler with various message contents."""
+ try:
+ dest = MagicMock()
+ dest.hash = os.urandom(16)
+ msg = LXMF.LXMessage(dest, mock_app.identity, content, title=title)
+
+ # Add random fields
+ msg.fields[LXMF.FIELD_TELEMETRY] = os.urandom(random.randint(0, 100))
+
+ # These are usually set by LXMF when receiving
+ msg.source_hash = os.urandom(16)
+ msg.hash = os.urandom(16)
+ msg.signature = os.urandom(64)
+
+ mock_app.on_lxmf_delivery(msg)
+ except Exception:
+ pass
+
+
+# Announce Ingestion Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ aspect=st.text(),
+ destination_hash=st.binary(min_size=0, max_size=64),
+ app_data=st.binary(min_size=0, max_size=1000),
+)
+def test_on_lxmf_announce_received_fuzzing(
+ mock_app, aspect, destination_hash, app_data
+):
+ """Fuzz the announce received handler."""
+ try:
+ announced_identity = MagicMock()
+ announced_identity.hash = os.urandom(32)
+ announced_identity.get_public_key = MagicMock(return_value=os.urandom(32))
+
+ mock_app.on_lxmf_announce_received(
+ aspect,
+ destination_hash,
+ announced_identity,
+ None, # reticulum instance
+ app_data,
+ )
+ except Exception:
+ pass
+
+
+# Interface Config Parser Fuzzing (More intensive)
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(config_text=st.text(min_size=0, max_size=10000))
+def test_interface_config_parser_deep_fuzzing(config_text):
+ """Deep fuzz the interface configuration parser."""
+ from meshchatx.src.backend.interface_config_parser import InterfaceConfigParser
+
+ try:
+ InterfaceConfigParser.parse(config_text)
+ except Exception:
+ # We don't care if it fails to parse, just that it doesn't crash the process
+ pass
+
+
+# Telemeter Roundtrip Fuzzing
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ battery=st.floats(0, 100, allow_nan=False, allow_infinity=False),
+ uptime=st.integers(0, 2**32 - 1),
+ load=st.floats(0, 100, allow_nan=False, allow_infinity=False),
+ temperature=st.floats(-50, 100, allow_nan=False, allow_infinity=False),
+)
+def test_telemeter_roundtrip_fuzzing(battery, uptime, load, temperature):
+ """Fuzz telemeter packing and unpacking in a roundtrip."""
+ from meshchatx.src.backend.telemetry_utils import Telemeter
+
+ try:
+ t = Telemeter(
+ battery=battery, uptime=uptime, load=load, temperature=temperature
+ )
+ packed = t.pack()
+ unpacked = Telemeter.from_packed(packed)
+ assert isinstance(unpacked, Telemeter)
+ except Exception:
+ pass
diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py
index cfddb355..f81934d7 100644
--- a/tests/backend/test_security_fuzzing.py
+++ b/tests/backend/test_security_fuzzing.py
@@ -1,4 +1,6 @@
import os
+import time
+import base64
from contextlib import ExitStack
from unittest.mock import MagicMock, patch
@@ -22,6 +24,7 @@ def mock_app():
self.hexhash = self.hash.hex()
with ExitStack() as stack:
+ # Mock core dependencies that interact with the system/network
stack.enter_context(patch("meshchatx.src.backend.identity_context.Database"))
stack.enter_context(
patch("meshchatx.src.backend.identity_context.ConfigManager")
@@ -58,40 +61,34 @@ def mock_app():
stack.enter_context(
patch("meshchatx.src.backend.identity_context.CommunityInterfacesManager")
)
+
mock_async_utils = stack.enter_context(patch("meshchatx.meshchat.AsyncUtils"))
stack.enter_context(patch("LXMF.LXMRouter"))
stack.enter_context(patch("RNS.Identity", MockIdentityClass))
stack.enter_context(patch("RNS.Reticulum"))
stack.enter_context(patch("RNS.Transport"))
mock_packet = stack.enter_context(patch("RNS.Packet"))
- mock_packet.MTU = 500 # Set a default MTU to avoid MagicMock comparison errors
- stack.enter_context(patch("threading.Thread"))
+ mock_packet.MTU = 500
+
+ # Stop background loops
stack.enter_context(
- patch.object(
- ReticulumMeshChat, "announce_loop", new=MagicMock(return_value=None)
- ),
+ patch.object(ReticulumMeshChat, "announce_loop", return_value=None)
)
stack.enter_context(
patch.object(
- ReticulumMeshChat,
- "announce_sync_propagation_nodes",
- new=MagicMock(return_value=None),
- ),
+ ReticulumMeshChat, "announce_sync_propagation_nodes", return_value=None
+ )
)
stack.enter_context(
- patch.object(
- ReticulumMeshChat, "crawler_loop", new=MagicMock(return_value=None)
- ),
+ patch.object(ReticulumMeshChat, "crawler_loop", return_value=None)
)
stack.enter_context(
- patch.object(
- ReticulumMeshChat, "auto_backup_loop", new=MagicMock(return_value=None)
- ),
+ patch.object(ReticulumMeshChat, "auto_backup_loop", return_value=None)
)
stack.enter_context(
patch.object(
ReticulumMeshChat, "send_config_to_websocket_clients", return_value=None
- ),
+ )
)
mock_id = MockIdentityClass()
@@ -107,7 +104,6 @@ def mock_app():
patch.object(MockIdentityClass, "from_bytes", return_value=mock_id)
)
- # Make run_async a no-op that doesn't trigger coroutine warnings
def mock_run_async(coro):
import asyncio
@@ -116,26 +112,29 @@ def mock_app():
mock_async_utils.run_async = MagicMock(side_effect=mock_run_async)
+ # Fix TelephoneManager.initiate to be awaitable
+ async def mock_initiate(*args, **kwargs):
+ return MagicMock()
+
+ mock_telephone_manager = stack.enter_context(
+ patch("meshchatx.src.backend.identity_context.TelephoneManager")
+ )
+ mock_telephone_manager.return_value.initiate = MagicMock(
+ side_effect=mock_initiate
+ )
+
app = ReticulumMeshChat(
identity=mock_id,
storage_dir="/tmp/meshchat_test",
reticulum_config_dir="/tmp/meshchat_test",
)
- # Setup config mock to return real values to avoid background thread issues
+ # Basic config setup
app.config = MagicMock()
app.config.auto_announce_enabled.get.return_value = False
- app.config.auto_announce_interval_seconds.get.return_value = 600
- app.config.last_announced_at.get.return_value = 0
- app.config.lxmf_auto_sync_propagation_nodes_enabled.get.return_value = False
- app.config.lxmf_auto_sync_propagation_nodes_interval_seconds.get.return_value = 3600
- app.config.lxmf_auto_sync_propagation_nodes_last_synced_at.get.return_value = 0
app.config.voicemail_enabled.get.return_value = True
- app.config.voicemail_auto_answer_delay_seconds.get.return_value = 0
- app.config.voicemail_greeting.get.return_value = "Hello"
- app.config.voicemail_max_recording_seconds.get.return_value = 10
- # Other required mocks for on_lxmf_delivery
+ # Surface mocks for tracking
app.is_destination_blocked = MagicMock(return_value=False)
app.check_spam_keywords = MagicMock(return_value=False)
app.db_upsert_lxmf_message = MagicMock()
@@ -143,6 +142,10 @@ def mock_app():
app.update_lxmf_user_icon = MagicMock()
app.websocket_broadcast = MagicMock()
+ # Avoid crashing during broadcast by returning None for message lookup
+ mock_db = app.current_context.database
+ mock_db.messages.get_lxmf_message_by_hash.return_value = None
+
yield app
app.teardown_identity()
@@ -179,15 +182,18 @@ def mock_app():
)
def test_lxmf_icon_appearance_fuzzing(mock_app, field_data):
"""Fuzz LXMF.FIELD_ICON_APPEARANCE parsing in on_lxmf_delivery."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+
mock_message = MagicMock()
mock_message.get_fields.return_value = {LXMF.FIELD_ICON_APPEARANCE: field_data}
mock_message.source_hash = os.urandom(16)
mock_message.hash = os.urandom(16)
- try:
- mock_app.on_lxmf_delivery(mock_message)
- except Exception:
- pass
+ # Should not crash even with malformed icon data
+ mock_app.on_lxmf_delivery(mock_message)
+
+ # Message should still be saved to DB regardless of icon parsing success
+ mock_app.db_upsert_lxmf_message.assert_called_once()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -216,6 +222,8 @@ def test_lxmf_icon_appearance_fuzzing(mock_app, field_data):
)
def test_lxmf_attachments_fuzzing(mock_app, attachments_data):
"""Fuzz LXMF.FIELD_FILE_ATTACHMENTS parsing."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+
mock_message = MagicMock()
mock_message.get_fields.return_value = {
LXMF.FIELD_FILE_ATTACHMENTS: attachments_data,
@@ -223,10 +231,8 @@ def test_lxmf_attachments_fuzzing(mock_app, attachments_data):
mock_message.source_hash = os.urandom(16)
mock_message.hash = os.urandom(16)
- try:
- mock_app.on_lxmf_delivery(mock_message)
- except Exception:
- pass
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -250,15 +256,15 @@ def test_lxmf_attachments_fuzzing(mock_app, attachments_data):
)
def test_lxmf_image_field_fuzzing(mock_app, image_data):
"""Fuzz LXMF.FIELD_IMAGE parsing."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+
mock_message = MagicMock()
mock_message.get_fields.return_value = {LXMF.FIELD_IMAGE: image_data}
mock_message.source_hash = os.urandom(16)
mock_message.hash = os.urandom(16)
- try:
- mock_app.on_lxmf_delivery(mock_message)
- except Exception:
- pass
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -282,15 +288,15 @@ def test_lxmf_image_field_fuzzing(mock_app, image_data):
)
def test_lxmf_audio_field_fuzzing(mock_app, audio_data):
"""Fuzz LXMF.FIELD_AUDIO parsing."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+
mock_message = MagicMock()
mock_message.get_fields.return_value = {LXMF.FIELD_AUDIO: audio_data}
mock_message.source_hash = os.urandom(16)
mock_message.hash = os.urandom(16)
- try:
- mock_app.on_lxmf_delivery(mock_message)
- except Exception:
- pass
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -299,7 +305,8 @@ def test_lxmf_audio_field_fuzzing(mock_app, audio_data):
file_bytes=st.binary(min_size=0, max_size=10000),
)
def test_attachment_filename_security(mock_app, filename, file_bytes):
- """Test for potential directory traversal or malicious filenames in attachments."""
+ """Test for potential crashes with malformed filenames in attachments."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
mock_message = MagicMock()
mock_message.get_fields.return_value = {
LXMF.FIELD_FILE_ATTACHMENTS: [[filename, file_bytes]],
@@ -307,26 +314,25 @@ def test_attachment_filename_security(mock_app, filename, file_bytes):
mock_message.source_hash = os.urandom(16)
mock_message.hash = os.urandom(16)
- try:
- mock_app.on_lxmf_delivery(mock_message)
- mock_app.convert_lxmf_message_to_dict(mock_message)
- except Exception:
- pass
+ # Should not crash
+ mock_app.on_lxmf_delivery(mock_message)
+ from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+ convert_lxmf_message_to_dict(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(caller_id_bytes=st.binary(min_size=0, max_size=1000))
def test_telephone_callback_fuzzing(mock_app, caller_id_bytes):
"""Fuzz telephone manager callbacks with malformed identity bytes."""
- try:
- mock_identity = MagicMock()
- mock_identity.hash = caller_id_bytes
+ mock_identity = MagicMock()
+ mock_identity.hash = caller_id_bytes
- mock_app.telephone_manager.on_telephone_ringing(mock_identity)
- mock_app.telephone_manager.on_telephone_call_established(mock_identity)
- mock_app.telephone_manager.on_telephone_call_ended(mock_identity)
- except Exception:
- pass
+ # Should handle malformed hashes gracefully without crashing
+ mock_app.telephone_manager.on_telephone_ringing(mock_identity)
+ mock_app.telephone_manager.on_telephone_call_established(mock_identity)
+ mock_app.telephone_manager.on_telephone_call_ended(mock_identity)
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -345,10 +351,8 @@ def test_telephone_callback_fuzzing(mock_app, caller_id_bytes):
)
def test_message_dao_upsert_fuzzing(mock_app, data):
"""Fuzz MessageDAO.upsert_lxmf_message with varied dictionary data."""
- try:
- mock_app.database.messages.upsert_lxmf_message(data)
- except Exception:
- pass
+ # This should not raise SQL errors or crash
+ mock_app.database.messages.upsert_lxmf_message(data)
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -374,10 +378,13 @@ def test_lxmf_message_decoding_fuzzing(mock_app, title_bytes, content_bytes):
mock_message.q = 100
mock_message.get_fields.return_value = {}
- try:
- mock_app.convert_lxmf_message_to_dict(mock_message)
- except Exception:
- pass
+ from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+ result = convert_lxmf_message_to_dict(mock_message)
+ assert isinstance(result, dict)
+ assert "hash" in result
+ assert "title" in result
+ assert "content" in result
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -389,9 +396,13 @@ def test_voicemail_greeting_fuzzing(mock_app, greeting_text):
mock_app.voicemail_manager.espeak_path = "/usr/bin/espeak"
mock_app.voicemail_manager.ffmpeg_path = "/usr/bin/ffmpeg"
- with patch("subprocess.run"):
+ with patch("subprocess.run") as mock_run:
+ mock_run.return_value = MagicMock(returncode=0)
try:
mock_app.voicemail_manager.generate_greeting(greeting_text)
+ # Should call subprocess to generate audio if text is not empty
+ if greeting_text.strip():
+ assert mock_run.called
except Exception:
pass
@@ -403,10 +414,8 @@ def test_voicemail_incoming_call_fuzzing(mock_app, caller_hash):
mock_identity = MagicMock()
mock_identity.hash = caller_hash
- try:
- mock_app.voicemail_manager.handle_incoming_call(mock_identity)
- except Exception:
- pass
+ # Should not crash with malformed identity hashes
+ mock_app.voicemail_manager.handle_incoming_call(mock_identity)
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -422,14 +431,12 @@ def test_forwarding_manager_mapping_fuzzing(
dest_hash,
):
"""Fuzz forwarding manager mapping creation."""
- try:
- mock_app.forwarding_manager.get_or_create_mapping(
- source_hash,
- recipient_hash,
- dest_hash,
- )
- except Exception:
- pass
+ # Should handle malformed hashes gracefully
+ mock_app.forwarding_manager.get_or_create_mapping(
+ source_hash,
+ recipient_hash,
+ dest_hash,
+ )
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -437,22 +444,20 @@ def test_forwarding_manager_mapping_fuzzing(
def test_lxm_ingest_uri_fuzzing(mock_app, uri):
"""Fuzz the lxm.ingest_uri WebSocket handler."""
mock_client = MagicMock()
- mock_client.send_str = MagicMock()
- try:
- # We need to wrap it in a task since it's async
- import asyncio
+ import asyncio
- loop = asyncio.new_event_loop()
- asyncio.set_event_loop(loop)
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
loop.run_until_complete(
mock_app.on_websocket_data_received(
mock_client,
{"type": "lxm.ingest_uri", "uri": uri},
),
)
- except Exception:
- pass
+ finally:
+ loop.close()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -471,20 +476,21 @@ def test_lxm_ingest_uri_fuzzing(mock_app, uri):
)
def test_update_config_fuzzing(mock_app, config_data):
"""Fuzz the update_config method with randomized dictionary data."""
- try:
- import asyncio
+ import asyncio
- loop = asyncio.new_event_loop()
- asyncio.set_event_loop(loop)
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
loop.run_until_complete(mock_app.update_config(config_data))
- except Exception:
- pass
+ finally:
+ loop.close()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(large_string=st.text(min_size=1000, max_size=10000))
def test_large_payload_dos_resistance(mock_app, large_string):
"""Check resistance to DoS via large strings in various fields."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
mock_message = MagicMock()
mock_message.title = large_string.encode()
mock_message.content = large_string.encode()
@@ -492,10 +498,9 @@ def test_large_payload_dos_resistance(mock_app, large_string):
mock_message.source_hash = os.urandom(16)
mock_message.get_fields.return_value = {}
- try:
- mock_app.on_lxmf_delivery(mock_message)
- except Exception:
- pass
+ # Should not crash or hang excessively
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -509,19 +514,19 @@ def test_large_payload_dos_resistance(mock_app, large_string):
def test_websocket_recursion_fuzzing(mock_app, nested_data):
"""Fuzz the WebSocket handler with deeply nested JSON data."""
mock_client = MagicMock()
- try:
- import asyncio
+ import asyncio
- loop = asyncio.new_event_loop()
- asyncio.set_event_loop(loop)
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
loop.run_until_complete(
mock_app.on_websocket_data_received(
mock_client,
{"type": "ping", "data": nested_data},
),
)
- except Exception:
- pass
+ finally:
+ loop.close()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -529,11 +534,11 @@ def test_websocket_recursion_fuzzing(mock_app, nested_data):
def test_lxm_generate_paper_uri_fuzzing(mock_app, dest_hash, content):
"""Fuzz paper URI generation with randomized inputs."""
mock_client = MagicMock()
- try:
- import asyncio
+ import asyncio
- loop = asyncio.new_event_loop()
- asyncio.set_event_loop(loop)
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
loop.run_until_complete(
mock_app.on_websocket_data_received(
mock_client,
@@ -544,8 +549,8 @@ def test_lxm_generate_paper_uri_fuzzing(mock_app, dest_hash, content):
},
),
)
- except Exception:
- pass
+ finally:
+ loop.close()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -556,10 +561,8 @@ def test_lxm_generate_paper_uri_fuzzing(mock_app, dest_hash, content):
)
def test_map_manager_coord_fuzzing(mock_app, lon, lat, zoom):
"""Fuzz coordinate to tile conversion in MapManager."""
- try:
- mock_app.map_manager._lonlat_to_tile(lon, lat, zoom)
- except Exception:
- pass
+ # Should handle invalid coordinates gracefully
+ mock_app.map_manager._lonlat_to_tile(lon, lat, zoom)
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -570,13 +573,9 @@ def test_map_manager_coord_fuzzing(mock_app, lon, lat, zoom):
)
def test_translator_handler_fuzzing(mock_app, text, source_lang, target_lang):
"""Fuzz the TranslatorHandler translate_text method."""
- try:
- # Mock dependencies
- mock_app.translator_handler.has_requests = False
- mock_app.translator_handler.has_argos = False
- mock_app.translator_handler.translate_text(text, source_lang, target_lang)
- except Exception:
- pass
+ mock_app.translator_handler.has_requests = False
+ mock_app.translator_handler.has_argos = False
+ mock_app.translator_handler.translate_text(text, source_lang, target_lang)
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -589,32 +588,1231 @@ def test_update_lxmf_user_icon_fuzzing(
bg_color,
):
"""Fuzz user icon update logic with malformed strings."""
- try:
- mock_app.update_lxmf_user_icon(dest_hash, icon_name, fg_color, bg_color)
- except Exception:
- pass
+ mock_app.update_lxmf_user_icon(dest_hash, icon_name, fg_color, bg_color)
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(binary_data=st.binary(min_size=0, max_size=10000))
def test_rns_identity_load_fuzzing(mock_app, binary_data):
"""Fuzz RNS.Identity loading with random binary data."""
+ # These should catch RNS internal errors if they are severe enough to crash Python
try:
- import RNS
+ RNS.Identity.from_bytes(binary_data)
+ except Exception:
+ pass
- try:
- RNS.Identity.from_bytes(binary_data)
- except Exception:
- pass
- try:
- id_inst = RNS.Identity(create_keys=False)
- id_inst.load_private_key(binary_data)
- except Exception:
- pass
- try:
- id_inst = RNS.Identity(create_keys=False)
- id_inst.load_public_key(binary_data)
- except Exception:
- pass
+ id_inst = RNS.Identity(create_keys=False)
+ try:
+ id_inst.load_private_key(binary_data)
+ except Exception:
+ pass
+
+ try:
+ id_inst.load_public_key(binary_data)
except Exception:
pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ uri=st.one_of(
+ st.text(min_size=0, max_size=10000),
+ st.binary(min_size=0, max_size=10000),
+ ),
+)
+def test_lxm_uri_comprehensive_fuzzing(mock_app, uri):
+ """Fuzz lxm:// and lxmf:// URI ingestion with various data types."""
+ mock_client = MagicMock()
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ uri_str = (
+ uri.decode("utf-8", errors="ignore") if isinstance(uri, bytes) else uri
+ )
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ mock_client,
+ {"type": "lxm.ingest_uri", "uri": uri_str},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ uri_prefix=st.sampled_from(["lxm://", "lxmf://", "LXM://", "LXMF://", "LxM://"]),
+ uri_body=st.text(min_size=0, max_size=5000),
+)
+def test_lxm_uri_prefix_variations_fuzzing(mock_app, uri_prefix, uri_body):
+ """Fuzz lxm:// URI with various prefix case combinations and malformed bodies."""
+ mock_client = MagicMock()
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ full_uri = uri_prefix + uri_body
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ mock_client,
+ {"type": "lxm.ingest_uri", "uri": full_uri},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(telemetry_packed=st.binary(min_size=0, max_size=10000))
+def test_telemetry_from_packed_fuzzing(mock_app, telemetry_packed):
+ """Fuzz Telemeter.from_packed with random binary data."""
+ from meshchatx.src.backend.telemetry_utils import Telemeter
+
+ Telemeter.from_packed(telemetry_packed)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ telemetry_data=st.one_of(
+ st.binary(min_size=0, max_size=10000),
+ st.text(min_size=0, max_size=1000),
+ st.lists(
+ st.one_of(
+ st.text(),
+ st.binary(),
+ st.integers(),
+ st.floats(),
+ st.booleans(),
+ st.none(),
+ ),
+ min_size=0,
+ max_size=10,
+ ),
+ st.dictionaries(
+ keys=st.text(),
+ values=st.one_of(
+ st.text(),
+ st.binary(),
+ st.integers(),
+ st.floats(),
+ st.booleans(),
+ st.none(),
+ ),
+ ),
+ ),
+)
+def test_lxmf_telemetry_field_fuzzing(mock_app, telemetry_data):
+ """Fuzz LXMF.FIELD_TELEMETRY parsing in on_lxmf_delivery."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.get_fields.return_value = {LXMF.FIELD_TELEMETRY: telemetry_data}
+ mock_message.source_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ icon_name=st.text(min_size=0, max_size=200),
+ foreground_bytes=st.one_of(
+ st.binary(min_size=0, max_size=100),
+ st.text(min_size=0, max_size=100),
+ st.integers(),
+ st.none(),
+ ),
+ background_bytes=st.one_of(
+ st.binary(min_size=0, max_size=100),
+ st.text(min_size=0, max_size=100),
+ st.integers(),
+ st.none(),
+ ),
+)
+def test_lxmf_icon_appearance_structure_fuzzing(
+ mock_app,
+ icon_name,
+ foreground_bytes,
+ background_bytes,
+):
+ """Fuzz LXMF.FIELD_ICON_APPEARANCE with structured data."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.get_fields.return_value = {
+ LXMF.FIELD_ICON_APPEARANCE: [icon_name, foreground_bytes, background_bytes],
+ }
+ mock_message.source_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ icon_appearance_list=st.lists(
+ st.one_of(
+ st.text(),
+ st.binary(),
+ st.integers(),
+ st.floats(),
+ st.booleans(),
+ st.none(),
+ ),
+ min_size=0,
+ max_size=10,
+ ),
+)
+def test_lxmf_icon_appearance_list_variations_fuzzing(mock_app, icon_appearance_list):
+ """Fuzz LXMF.FIELD_ICON_APPEARANCE with various list structures."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.get_fields.return_value = {
+ LXMF.FIELD_ICON_APPEARANCE: icon_appearance_list,
+ }
+ mock_message.source_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ dest_hash=st.text(min_size=0, max_size=100),
+ icon_name=st.text(min_size=0, max_size=500),
+ fg_color=st.text(min_size=0, max_size=100),
+ bg_color=st.text(min_size=0, max_size=100),
+)
+def test_update_lxmf_user_icon_comprehensive_fuzzing(
+ mock_app,
+ dest_hash,
+ icon_name,
+ fg_color,
+ bg_color,
+):
+ """Fuzz update_lxmf_user_icon with various string inputs."""
+ mock_app.update_lxmf_user_icon(dest_hash, icon_name, fg_color, bg_color)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ audio_mode=st.one_of(
+ st.text(min_size=0, max_size=100),
+ st.binary(min_size=0, max_size=100),
+ st.integers(),
+ st.floats(),
+ st.booleans(),
+ st.none(),
+ ),
+ audio_bytes=st.one_of(
+ st.binary(min_size=0, max_size=100000),
+ st.text(min_size=0, max_size=10000),
+ st.lists(st.integers(min_value=0, max_value=255), min_size=0, max_size=1000),
+ st.none(),
+ ),
+)
+def test_lxmf_audio_field_structure_fuzzing(mock_app, audio_mode, audio_bytes):
+ """Fuzz LXMF.FIELD_AUDIO with structured data."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.get_fields.return_value = {
+ LXMF.FIELD_AUDIO: [audio_mode, audio_bytes],
+ }
+ mock_message.source_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ audio_field=st.one_of(
+ st.lists(
+ st.one_of(
+ st.text(),
+ st.binary(),
+ st.integers(),
+ st.floats(),
+ st.booleans(),
+ st.none(),
+ ),
+ min_size=0,
+ max_size=20,
+ ),
+ st.binary(min_size=0, max_size=100000),
+ st.text(min_size=0, max_size=10000),
+ st.dictionaries(
+ keys=st.text(),
+ values=st.one_of(st.text(), st.binary(), st.integers()),
+ ),
+ ),
+)
+def test_lxmf_audio_field_variations_fuzzing(mock_app, audio_field):
+ """Fuzz LXMF.FIELD_AUDIO with various data structures."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.get_fields.return_value = {LXMF.FIELD_AUDIO: audio_field}
+ mock_message.source_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ contact_name=st.text(min_size=0, max_size=500),
+ contact_hash=st.text(min_size=0, max_size=100),
+)
+def test_contact_sharing_content_fuzzing(mock_app, contact_name, contact_hash):
+ """Fuzz contact sharing content parsing."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ contact_content = f"Contact: {contact_name} <{contact_hash}>"
+ mock_message = MagicMock()
+ mock_message.content = contact_content.encode("utf-8", errors="ignore")
+ mock_message.title = b""
+ mock_message.hash = os.urandom(16)
+ mock_message.source_hash = os.urandom(16)
+ mock_message.destination_hash = os.urandom(16)
+ mock_message.incoming = True
+ mock_message.state = LXMF.LXMessage.DELIVERED
+ mock_message.method = LXMF.LXMessage.DIRECT
+ mock_message.progress = 1.0
+ mock_message.timestamp = 123456789.0
+ mock_message.rssi = -50
+ mock_message.snr = 10
+ mock_message.q = 100
+ mock_message.get_fields.return_value = {}
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ contact_content=st.text(min_size=0, max_size=2000),
+)
+def test_contact_sharing_malformed_content_fuzzing(mock_app, contact_content):
+ """Fuzz contact sharing with malformed content strings."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.content = contact_content.encode("utf-8", errors="ignore")
+ mock_message.title = b""
+ mock_message.hash = os.urandom(16)
+ mock_message.source_hash = os.urandom(16)
+ mock_message.destination_hash = os.urandom(16)
+ mock_message.incoming = True
+ mock_message.state = LXMF.LXMessage.DELIVERED
+ mock_message.method = LXMF.LXMessage.DIRECT
+ mock_message.progress = 1.0
+ mock_message.timestamp = 123456789.0
+ mock_message.rssi = -50
+ mock_message.snr = 10
+ mock_message.q = 100
+ mock_message.get_fields.return_value = {}
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ name=st.text(min_size=0, max_size=500),
+ hash_str=st.text(min_size=0, max_size=100),
+)
+def test_add_contact_api_fuzzing(mock_app, name, hash_str):
+ """Fuzz contact addition API with various inputs."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {
+ "type": "telephone.add_contact",
+ "name": name,
+ "remote_identity_hash": hash_str,
+ },
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ time_utc=st.one_of(
+ st.integers(min_value=-2147483648, max_value=2147483647),
+ st.floats(min_value=-2147483648, max_value=2147483647),
+ st.none(),
+ ),
+ location=st.one_of(
+ st.fixed_dictionaries(
+ {
+ "latitude": st.floats(allow_nan=False, allow_infinity=False),
+ "longitude": st.floats(allow_nan=False, allow_infinity=False),
+ },
+ optional={
+ "altitude": st.floats(allow_nan=False, allow_infinity=False),
+ "speed": st.floats(allow_nan=False, allow_infinity=False),
+ "bearing": st.floats(allow_nan=False, allow_infinity=False),
+ "accuracy": st.floats(allow_nan=False, allow_infinity=False),
+ "last_update": st.floats(allow_nan=False, allow_infinity=False),
+ },
+ ),
+ st.none(),
+ ),
+)
+def test_telemetry_pack_fuzzing(mock_app, time_utc, location):
+ """Fuzz Telemeter.pack with various data."""
+ from meshchatx.src.backend.telemetry_utils import Telemeter
+
+ try:
+ Telemeter.pack(time_utc=time_utc, location=location)
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ packed_location=st.one_of(
+ st.lists(
+ st.one_of(
+ st.binary(min_size=0, max_size=100),
+ st.integers(),
+ st.text(),
+ st.floats(),
+ st.none(),
+ ),
+ min_size=0,
+ max_size=20,
+ ),
+ st.binary(min_size=0, max_size=1000),
+ st.text(min_size=0, max_size=1000),
+ st.none(),
+ ),
+)
+def test_telemetry_unpack_location_fuzzing(mock_app, packed_location):
+ """Fuzz Telemeter.unpack_location with various formats."""
+ from meshchatx.src.backend.telemetry_utils import Telemeter
+
+ Telemeter.unpack_location(packed_location)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ latitude=st.floats(allow_nan=True, allow_infinity=True),
+ longitude=st.floats(allow_nan=True, allow_infinity=True),
+ altitude=st.one_of(st.floats(allow_nan=True, allow_infinity=True), st.integers()),
+ speed=st.one_of(st.floats(allow_nan=True, allow_infinity=True), st.integers()),
+ bearing=st.one_of(st.floats(allow_nan=True, allow_infinity=True), st.integers()),
+ accuracy=st.one_of(st.floats(allow_nan=True, allow_infinity=True), st.integers()),
+ last_update=st.one_of(
+ st.integers(), st.floats(), st.text(), st.binary(), st.none()
+ ),
+)
+def test_telemetry_pack_location_fuzzing(
+ mock_app,
+ latitude,
+ longitude,
+ altitude,
+ speed,
+ bearing,
+ accuracy,
+ last_update,
+):
+ """Fuzz Telemeter.pack_location with edge case coordinates."""
+ from meshchatx.src.backend.telemetry_utils import Telemeter
+
+ Telemeter.pack_location(
+ latitude=latitude,
+ longitude=longitude,
+ altitude=altitude,
+ speed=speed,
+ bearing=bearing,
+ accuracy=accuracy,
+ last_update=last_update,
+ )
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ destination_hash=st.one_of(st.text(), st.binary()),
+ timestamp=st.one_of(
+ st.integers(),
+ st.floats(allow_nan=False, allow_infinity=False),
+ st.text(),
+ st.none(),
+ ),
+ data=st.one_of(
+ st.text(), st.binary(), st.dictionaries(keys=st.text(), values=st.text())
+ ),
+ received_from=st.one_of(st.text(), st.binary(), st.none()),
+ physical_link=st.one_of(
+ st.dictionaries(
+ keys=st.text(), values=st.one_of(st.integers(), st.floats(), st.text())
+ ),
+ st.text(),
+ st.binary(),
+ st.none(),
+ ),
+)
+def test_telemetry_upsert_fuzzing(
+ mock_app,
+ destination_hash,
+ timestamp,
+ data,
+ received_from,
+ physical_link,
+):
+ """Fuzz telemetry database upsert with varied data types."""
+ dest_hash_str = (
+ destination_hash.decode("utf-8", errors="ignore")
+ if isinstance(destination_hash, bytes)
+ else str(destination_hash)
+ )
+ mock_app.database.telemetry.upsert_telemetry(
+ destination_hash=dest_hash_str,
+ timestamp=timestamp,
+ data=data,
+ received_from=received_from,
+ physical_link=physical_link,
+ )
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ z=st.one_of(st.integers(), st.text(), st.floats()),
+ x=st.one_of(st.integers(), st.text(), st.floats()),
+ y=st.one_of(st.integers(), st.text(), st.floats()),
+)
+def test_map_tile_coordinates_fuzzing(mock_app, z, x, y):
+ """Fuzz map tile coordinate parsing."""
+ try:
+ z_int = (
+ int(z)
+ if isinstance(z, (int, float))
+ and not (isinstance(z, float) and (z != z or abs(z) == float("inf")))
+ else 0
+ )
+ x_int = (
+ int(x)
+ if isinstance(x, (int, float))
+ and not (isinstance(x, float) and (x != x or abs(x) == float("inf")))
+ else 0
+ )
+ y_int = (
+ int(y)
+ if isinstance(y, (int, float))
+ and not (isinstance(y, float) and (y != y or abs(y) == float("inf")))
+ else 0
+ )
+ mock_app.map_manager.get_tile(z_int, x_int, y_int)
+ except (OverflowError, ValueError):
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ filename=st.text(min_size=0, max_size=500),
+)
+def test_mbtiles_filename_fuzzing(mock_app, filename):
+ """Fuzz MBTiles filename handling."""
+ mock_app.map_manager.delete_mbtiles(filename)
+ mock_app.map_manager.get_connection(filename)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ destination_hash=st.text(min_size=0, max_size=100),
+ page_path=st.text(min_size=0, max_size=1000),
+ content=st.text(min_size=0, max_size=100000),
+)
+def test_archive_page_content_fuzzing(mock_app, destination_hash, page_path, content):
+ """Fuzz archive page content storage and retrieval."""
+ mock_app.archiver_manager.archive_page(
+ destination_hash,
+ page_path,
+ content,
+ max_versions=5,
+ max_storage_gb=1,
+ )
+ mock_app.archiver_manager.get_archived_page_versions(destination_hash, page_path)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ ids=st.lists(
+ st.one_of(st.integers(), st.text(), st.floats()),
+ min_size=0,
+ max_size=100,
+ ),
+)
+def test_delete_archived_pages_ids_fuzzing(mock_app, ids):
+ """Fuzz SQL injection in delete_archived_pages."""
+ mock_app.database.misc.delete_archived_pages(ids=ids)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ query=st.text(min_size=0, max_size=500),
+)
+def test_archived_pages_query_sql_injection_fuzzing(mock_app, query):
+ """Fuzz SQL injection in archived_pages search."""
+ mock_app.database.misc.get_archived_pages_paginated(query=query)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ file_path=st.text(min_size=0, max_size=1000),
+)
+def test_rncp_file_path_traversal_fuzzing(mock_app, file_path):
+ """Fuzz RNCP file path handling for directory traversal."""
+ try:
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ loop.run_until_complete(
+ mock_app.rncp_handler.send_file(
+ os.urandom(16),
+ file_path,
+ timeout=1.0,
+ ),
+ )
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ path=st.text(min_size=0, max_size=1000),
+ data=st.one_of(st.text(), st.binary()),
+ request_id=st.one_of(st.integers(), st.text()),
+)
+def test_rncp_fetch_request_path_fuzzing(mock_app, path, data, request_id):
+ """Fuzz RNCP fetch request path handling."""
+ try:
+ mock_identity = MagicMock()
+ mock_identity.hash = os.urandom(16)
+ mock_app.rncp_handler._fetch_request(
+ path,
+ data,
+ request_id,
+ os.urandom(16),
+ mock_identity,
+ time.time(),
+ )
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ app_data_base64=st.text(min_size=0, max_size=10000),
+)
+def test_parse_lxmf_stamp_cost_fuzzing(mock_app, app_data_base64):
+ """Fuzz LXMF stamp cost parsing from base64 app_data."""
+ try:
+ from meshchatx.src.backend.meshchat_utils import parse_lxmf_stamp_cost
+
+ parse_lxmf_stamp_cost(app_data_base64)
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ app_data_base64=st.text(min_size=0, max_size=10000),
+)
+def test_parse_lxmf_propagation_node_app_data_fuzzing(mock_app, app_data_base64):
+ """Fuzz LXMF propagation node app_data parsing."""
+ from meshchatx.src.backend.meshchat_utils import (
+ parse_lxmf_propagation_node_app_data,
+ )
+
+ parse_lxmf_propagation_node_app_data(app_data_base64)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ destination_hash=st.text(min_size=0, max_size=100),
+ page_path=st.text(min_size=0, max_size=1000),
+)
+def test_nomadnet_page_path_fuzzing(mock_app, destination_hash, page_path):
+ """Fuzz NomadNet page path handling."""
+ mock_app.nomadnet_manager.archive_page(
+ destination_hash,
+ page_path,
+ "test content",
+ is_manual=False,
+ )
+ mock_app.nomadnet_manager.get_archived_page_versions(destination_hash, page_path)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ page_content=st.text(min_size=0, max_size=100000),
+)
+def test_nomadnet_page_content_fuzzing(mock_app, page_content):
+ """Fuzz NomadNet page content parsing."""
+ from meshchatx.src.backend.nomadnet_utils import (
+ convert_nomadnet_field_data_to_map,
+ )
+
+ convert_nomadnet_field_data_to_map(page_content)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ table_name=st.text(min_size=0, max_size=100),
+)
+def test_sql_table_name_injection_fuzzing(mock_app, table_name):
+ """Fuzz SQL table name injection."""
+ mock_app.database.provider.execute(f"PRAGMA table_info({table_name})")
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ fields_json=st.text(min_size=0, max_size=10000),
+)
+def test_lxmf_fields_json_parsing_fuzzing(mock_app, fields_json):
+ """Fuzz LXMF fields JSON parsing."""
+ db_message = {
+ "id": 1,
+ "hash": "test",
+ "source_hash": "test",
+ "destination_hash": "test",
+ "is_incoming": True,
+ "state": "delivered",
+ "progress": 100.0,
+ "method": "direct",
+ "delivery_attempts": 0,
+ "next_delivery_attempt_at": None,
+ "title": "test",
+ "content": "test",
+ "fields": fields_json,
+ "timestamp": 123456789.0,
+ "rssi": -50,
+ "snr": 10,
+ "quality": 100,
+ "is_spam": False,
+ "created_at": "2024-01-01T00:00:00Z",
+ "updated_at": "2024-01-01T00:00:00Z",
+ }
+ from meshchatx.src.backend.lxmf_utils import convert_db_lxmf_message_to_dict
+
+ try:
+ convert_db_lxmf_message_to_dict(db_message)
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ password=st.text(min_size=0, max_size=1000),
+)
+def test_auth_password_fuzzing(mock_app, password):
+ """Fuzz authentication password handling."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {"type": "auth.login", "password": password},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ filename=st.text(min_size=0, max_size=500),
+)
+def test_mbtiles_upload_filename_fuzzing(mock_app, filename):
+ """Fuzz MBTiles upload filename."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ mock_field = MagicMock()
+ mock_field.name = "file"
+ mock_field.filename = filename
+ mock_field.read_chunk = MagicMock(return_value=b"")
+ mock_reader = MagicMock()
+ mock_reader.next = MagicMock(return_value=mock_field)
+
+ mock_request = MagicMock()
+ mock_request.multipart = MagicMock(return_value=mock_reader)
+
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {"type": "map.upload_offline", "filename": filename},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ crawl_destination_hash=st.text(min_size=0, max_size=100),
+ crawl_page_path=st.text(min_size=0, max_size=1000),
+)
+def test_crawler_task_path_fuzzing(mock_app, crawl_destination_hash, crawl_page_path):
+ """Fuzz crawler task destination hash and page path."""
+ mock_app.database.misc.upsert_crawl_task(
+ crawl_destination_hash,
+ crawl_page_path,
+ status="pending",
+ retry_count=0,
+ )
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ doc_path=st.text(min_size=0, max_size=1000),
+)
+def test_docs_path_traversal_fuzzing(mock_app, doc_path):
+ """Fuzz documentation path handling."""
+ try:
+ mock_app.docs_manager.get_doc_content(doc_path)
+ except (IsADirectoryError, OSError):
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ bbox=st.lists(
+ st.one_of(st.floats(allow_nan=False, allow_infinity=False), st.integers()),
+ min_size=4,
+ max_size=4,
+ ),
+ min_zoom=st.integers(min_value=-10, max_value=30),
+ max_zoom=st.integers(min_value=-10, max_value=30),
+ name=st.text(min_size=0, max_size=500),
+)
+def test_map_export_parameters_fuzzing(mock_app, bbox, min_zoom, max_zoom, name):
+ """Fuzz map export parameters."""
+ mock_app.map_manager.start_export(
+ "test_export",
+ bbox,
+ min_zoom,
+ max_zoom,
+ name=name,
+ )
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ mbtiles_path=st.text(min_size=0, max_size=1000),
+)
+def test_mbtiles_metadata_parsing_fuzzing(mock_app, mbtiles_path):
+ """Fuzz MBTiles metadata parsing."""
+ mock_app.map_manager.get_metadata()
+ if os.path.exists(mbtiles_path):
+ mock_app.map_manager.get_connection(mbtiles_path)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ audio_frame=st.one_of(
+ st.binary(min_size=0, max_size=10000),
+ st.lists(st.integers(min_value=0, max_value=255), min_size=0, max_size=1000),
+ st.text(min_size=0, max_size=1000),
+ st.none(),
+ ),
+)
+def test_lxst_audio_frame_handling_fuzzing(mock_app, audio_frame):
+ """Fuzz LXST audio frame handling in Tee.handle_frame."""
+ try:
+ from meshchatx.src.backend.telephone_manager import Tee
+
+ mock_sink = MagicMock()
+ mock_sink.handle_frame = MagicMock()
+ mock_sink.can_receive = MagicMock(return_value=True)
+ tee = Tee(mock_sink)
+ tee.handle_frame(audio_frame, "test_source")
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ call_status=st.integers(min_value=-10, max_value=20),
+ caller_identity_hash=st.binary(min_size=0, max_size=100),
+)
+def test_lxst_call_state_transitions_fuzzing(
+ mock_app, call_status, caller_identity_hash
+):
+ """Fuzz LXST call state transitions with invalid states."""
+ try:
+ mock_identity = MagicMock()
+ mock_identity.hash = caller_identity_hash
+
+ if (
+ hasattr(mock_app.telephone_manager, "telephone")
+ and mock_app.telephone_manager.telephone
+ ):
+ mock_app.telephone_manager.telephone.call_status = call_status
+ mock_app.telephone_manager.on_telephone_ringing(mock_identity)
+ mock_app.telephone_manager.on_telephone_call_established(mock_identity)
+ mock_app.telephone_manager.on_telephone_call_ended(mock_identity)
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ codec2_data=st.binary(min_size=0, max_size=100000),
+ codec_mode=st.sampled_from(
+ [
+ "450PWB",
+ "450",
+ "700C",
+ "1200",
+ "1300",
+ "1400",
+ "1600",
+ "2400",
+ "3200",
+ "invalid",
+ ]
+ ),
+)
+def test_codec2_decode_fuzzing(mock_app, codec2_data, codec_mode):
+ """Fuzz Codec2 audio decoding with malformed data."""
+ try:
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+
+ mock_client = MagicMock()
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ mock_client,
+ {
+ "type": "codec2.decode",
+ "data": codec2_data,
+ "mode": codec_mode,
+ },
+ ),
+ )
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ opus_data=st.one_of(
+ st.binary(min_size=0, max_size=100000),
+ st.text(min_size=0, max_size=10000),
+ st.lists(st.integers(min_value=0, max_value=255), min_size=0, max_size=10000),
+ ),
+)
+def test_opus_audio_decode_fuzzing(mock_app, opus_data):
+ """Fuzz Opus audio decoding in LXMF audio fields."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.get_fields.return_value = {
+ LXMF.FIELD_AUDIO: [0x10, opus_data], # AM_OPUS_OGG = 0x10
+ }
+ mock_message.source_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ audio_mode=st.integers(min_value=0, max_value=255),
+ audio_bytes=st.binary(min_size=0, max_size=100000),
+)
+def test_lxmf_audio_mode_fuzzing(mock_app, audio_mode, audio_bytes):
+ """Fuzz all possible LXMF audio mode values."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.get_fields.return_value = {
+ LXMF.FIELD_AUDIO: [audio_mode, audio_bytes],
+ }
+ mock_message.source_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ profile_id=st.one_of(
+ st.integers(min_value=-100, max_value=100),
+ st.text(min_size=0, max_size=100),
+ st.none(),
+ ),
+)
+def test_lxst_profile_switching_fuzzing(mock_app, profile_id):
+ """Fuzz LXST audio profile switching."""
+ if (
+ hasattr(mock_app.telephone_manager, "telephone")
+ and mock_app.telephone_manager.telephone
+ ):
+ mock_app.telephone_manager.telephone.switch_profile(profile_id)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ destination_hash=st.one_of(
+ st.binary(min_size=0, max_size=100), st.text(min_size=0, max_size=100)
+ ),
+ timeout=st.one_of(
+ st.integers(min_value=-100, max_value=1000),
+ st.floats(allow_nan=True, allow_infinity=True),
+ ),
+)
+def test_lxst_call_initiation_fuzzing(mock_app, destination_hash, timeout):
+ """Fuzz LXST call initiation."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ dest_hash_bytes = (
+ bytes.fromhex(destination_hash)
+ if isinstance(destination_hash, str) and len(destination_hash) == 32
+ else destination_hash
+ if isinstance(destination_hash, bytes)
+ else os.urandom(16)
+ )
+ timeout_int = (
+ int(timeout)
+ if isinstance(timeout, (int, float))
+ and not (
+ isinstance(timeout, float)
+ and (timeout != timeout or abs(timeout) == float("inf"))
+ )
+ else 15
+ )
+
+ loop.run_until_complete(
+ mock_app.telephone_manager.initiate(
+ dest_hash_bytes, timeout_seconds=timeout_int
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ micron_content=st.text(min_size=0, max_size=50000),
+)
+def test_micron_parser_content_fuzzing(mock_app, micron_content):
+ """Fuzz Micron parser content handling."""
+ mock_app.db_upsert_lxmf_message.reset_mock()
+ mock_message = MagicMock()
+ mock_message.content = micron_content.encode("utf-8", errors="ignore")
+ mock_message.title = b""
+ mock_message.hash = os.urandom(16)
+ mock_message.source_hash = os.urandom(16)
+ mock_message.destination_hash = os.urandom(16)
+ mock_message.incoming = True
+ mock_message.state = LXMF.LXMessage.DELIVERED
+ mock_message.method = LXMF.LXMessage.DIRECT
+ mock_message.progress = 1.0
+ mock_message.timestamp = 123456789.0
+ mock_message.rssi = -50
+ mock_message.snr = 10
+ mock_message.q = 100
+ mock_message.get_fields.return_value = {}
+
+ mock_app.on_lxmf_delivery(mock_message)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ voicemail_text=st.text(min_size=0, max_size=10000),
+)
+def test_voicemail_greeting_text_fuzzing(mock_app, voicemail_text):
+ """Fuzz voicemail greeting generation."""
+ from meshchatx.src.backend.voicemail_manager import VoicemailManager
+
+ # Use real VoicemailManager to test its internal logic calling subprocess
+ vm = VoicemailManager(MagicMock(), MagicMock(), MagicMock(), "/tmp/voicemail_test")
+ vm.has_espeak = True
+ vm.has_ffmpeg = True
+ vm.espeak_path = "/usr/bin/espeak"
+ vm.ffmpeg_path = "/usr/bin/ffmpeg"
+
+ with patch("subprocess.run") as mock_run:
+ mock_run.return_value = MagicMock(returncode=0)
+ try:
+ vm.generate_greeting(voicemail_text)
+ # If text is provided, it should call subprocess.run
+ if voicemail_text.strip():
+ assert mock_run.called
+ except Exception:
+ # Ignore errors from underlying tools in fuzzing
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ rns_packet_data=st.binary(min_size=0, max_size=10000),
+)
+def test_rns_packet_parsing_fuzzing(mock_app, rns_packet_data):
+ """Fuzz RNS packet parsing with malformed protocol data."""
+ try:
+ import RNS
+
+ try:
+ RNS.Packet(None, rns_packet_data)
+ except Exception:
+ pass
+ try:
+ RNS.Packet.unpack(rns_packet_data)
+ except Exception:
+ pass
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ lxmf_message_data=st.binary(min_size=0, max_size=100000),
+)
+def test_lxmf_message_unpacking_fuzzing(mock_app, lxmf_message_data):
+ """Fuzz LXMF message unpacking."""
+ try:
+ LXMF.LXMessage.unpack(lxmf_message_data)
+ except Exception:
+ pass
+ try:
+ message = LXMF.LXMessage(None, None, "")
+ message.unpack(lxmf_message_data)
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ pipeline_config=st.dictionaries(
+ keys=st.text(),
+ values=st.one_of(
+ st.text(), st.binary(), st.integers(), st.floats(), st.booleans(), st.none()
+ ),
+ ),
+)
+def test_lxst_pipeline_config_fuzzing(mock_app, pipeline_config):
+ """Fuzz LXST Pipeline configuration."""
+ from LXST.Pipeline import Pipeline
+ from LXST.Codecs import Null
+ from LXST.Sources import Source
+ from LXST.Sinks import Sink
+
+ class DummySource(Source):
+ pass
+
+ class DummySink(Sink):
+ pass
+
+ # Pipeline requires source, codec, and sink
+ try:
+ pipeline = Pipeline(source=DummySource(), codec=Null(), sink=DummySink())
+ for key, value in pipeline_config.items():
+ try:
+ setattr(pipeline, key, value)
+ except Exception:
+ pass
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ sink_data=st.one_of(
+ st.binary(min_size=0, max_size=10000), st.text(min_size=0, max_size=1000)
+ ),
+)
+def test_lxst_sink_handling_fuzzing(mock_app, sink_data):
+ """Fuzz LXST sink data handling."""
+ from LXST.Sinks import OpusFileSink
+
+ sink = OpusFileSink("/tmp/test.opus")
+ try:
+ sink.handle_frame(sink_data, "test_source")
+ except Exception:
+ pass
+ try:
+ sink.can_receive("test_source")
+ except Exception:
+ pass
+
+
+def test_telemetry_packing_invariants_regression():
+ """Deterministic regression test for telemetry packing/unpacking."""
+ from meshchatx.src.backend.telemetry_utils import Telemeter
+
+ original_data = {
+ "time": {"utc": 123456789.0},
+ "location": {
+ "latitude": 45.0,
+ "longitude": -90.0,
+ "altitude": 100,
+ "speed": 10,
+ "bearing": 180,
+ "accuracy": 5,
+ "last_update": 123456780.0,
+ },
+ }
+
+ packed = Telemeter.pack(
+ time_utc=original_data["time"]["utc"], location=original_data["location"]
+ )
+ unpacked = Telemeter.from_packed(packed)
+
+ assert unpacked["time"]["utc"] == original_data["time"]["utc"]
+ assert unpacked["location"]["latitude"] == original_data["location"]["latitude"]
+ assert unpacked["location"]["longitude"] == original_data["location"]["longitude"]
+
+
+def test_lxmf_display_name_parsing_regression():
+ """Deterministic regression test for LXMF display name parsing."""
+ from meshchatx.src.backend.meshchat_utils import parse_lxmf_display_name
+
+ valid_b64 = base64.b64encode(b"test").decode()
+
+ with patch("LXMF.display_name_from_app_data") as mock_parser:
+ # Success case
+ mock_parser.return_value = "Test User"
+ assert parse_lxmf_display_name(valid_b64) == "Test User"
+
+ # None case (fallback to default)
+ mock_parser.return_value = None
+ assert (
+ parse_lxmf_display_name(valid_b64, default_value="Fallback") == "Fallback"
+ )
+
+ # Exception case
+ mock_parser.side_effect = Exception("Parsing error")
+ assert (
+ parse_lxmf_display_name(valid_b64, default_value="Fallback") == "Fallback"
+ )
+
+ # None input
+ assert parse_lxmf_display_name(None, default_value="Fallback") == "Fallback"
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────